package com.ibm.cps.storm;

import java.util.ArrayList;
import java.util.Collection;
import java.util.HashMap;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Set;

import org.apache.log4j.Logger;

import com.google.common.base.Throwables;
import com.google.gson.Gson;
import com.ibm.cps.message.DataMessage;
import com.ibm.cps.message.OptimizedParsedDataMessage;
import com.ibm.cps.message.ParsedDataMessage;
import com.ibm.cps.newmessage.AbstractMetadata;
import com.ibm.cps.newmessage.MetadataFactory;
import com.ibm.util.ErrorCode;
import com.ibm.util.TopologyMessageFieldIds;
import com.ibm.util.TopologyStreamIds;
import com.ibm.util.exception.CPSException;

import backtype.storm.task.OutputCollector;
import backtype.storm.task.TopologyContext;
import backtype.storm.topology.OutputFieldsDeclarer;
import backtype.storm.topology.base.BaseRichBolt;
import backtype.storm.tuple.Fields;
import backtype.storm.tuple.Tuple;
import backtype.storm.tuple.Values;

public class DispatchBolt extends BaseRichBolt {

    /**
     *
     */
    private static final long serialVersionUID = 1L;
    private HashMap<String, Set<String>> processor2successors;
    private OutputCollector collector;
    private static final Logger logger = Logger.getLogger(DispatchBolt.class);
    ;

    @Override
    public void prepare(Map stormConf, TopologyContext context, OutputCollector collector) {
        processor2successors = new HashMap<String, Set<String>>();
        this.collector = collector;
    }

    @Override
    public void execute(Tuple input) {
        try {
            String sourceStream = input.getSourceStreamId();
            logger.debug("source stream is " + sourceStream);
            if (isDataMessageStream(sourceStream) || isBatchDataMessageStream(sourceStream)) {
                Object message = input.getValue(0);
                Object tsKey = input.getValue(1);
                String source = input.getString(2);
                if (isDataMessageStream(sourceStream)) {
                    String parentId = getProcessorId((DataMessage) message, source);
                    sendMessage2Successors(message, parentId, tsKey);
                } else {
                    groupMessageBySuccessor(message, tsKey, source);
                }
            } else if (isMetaDataStream(sourceStream)) {
                AbstractMetadata metadata = (AbstractMetadata) input.getValue(0);
                MetadataFactory.updateProcessorMap(processor2successors, metadata);
                collector.emit(TopologyStreamIds.METADATA_STREAM, new Values(metadata));
                logger.debug("After add processor " + metadata.getProcessorid() + ", current map is"
                        + new Gson().toJson(processor2successors));
            } else if (idDeleteMetadataStream(sourceStream)) {
                logger.debug("Processor map before delete is " + new Gson().toJson(processor2successors));
                if (processor2successors != null) {
                    AbstractMetadata metaMessage = (AbstractMetadata) input.getValue(0);
                    List<List<Object>> retMsg = MetadataFactory.deleteProcessor(processor2successors, metaMessage);
                    logger.debug("Rest processor map is: " + new Gson().toJson(processor2successors));
                    if (retMsg != null) {
                        for (List<Object> tuple : retMsg) {
                            collector.emit(TopologyStreamIds.METADATA_DELETE_STREAM, tuple);
                        }
                    }
                }
            }
        } catch (Throwable e) {
            logger.error(Throwables.getStackTraceAsString(e));
            logger.error(e.getMessage());
        }
    }

    private void groupMessageBySuccessor(Object message, Object tskey, String posProcessorid) throws CPSException {
        if (!(message instanceof Collection)) {
            throw new CPSException(ErrorCode.INVALID_INPUT_ERROR,
                    "Message in batch stream should be instance of Collection.");
        }
        Collection<DataMessage> msgs = (Collection<DataMessage>) message;
        Map<String, Collection<DataMessage>> successor2Messages = new HashMap<String, Collection<DataMessage>>();
        String processorid;
        if (msgs != null) {

            // group message by successors
            for (DataMessage msg : msgs) {
                processorid = getProcessorId(msg, posProcessorid);
                Set<String> successors = processor2successors.get(processorid);
                if (successors != null) {
                    for (String successor : successors) {
                        if (successor2Messages.get(successor) == null) {
                            successor2Messages.put(successor, new ArrayList<DataMessage>());
                        }
                        successor2Messages.get(successor).add(msg);
                    }
                }
            }

            // send message by group
            for (Map.Entry<String, Collection<DataMessage>> entry : successor2Messages.entrySet()) {
                Iterator<DataMessage> ite = entry.getValue().iterator();
                while (ite.hasNext()) {
                    DataMessage msg = ite.next();

                    // not end
                    if (ite.hasNext()) {
                        collector.emit(TopologyStreamIds.DATAMESSAGE_STREAM,
                                new Values(entry.getKey(), msg, tskey, true));
                    } else {
                        collector.emit(TopologyStreamIds.DATAMESSAGE_STREAM,
                                new Values(entry.getKey(), msg, tskey, false));
                    }
                }

            }

        }
    }

    /**
     * 1 to 1 processors
     *
     * @param message
     * @param parentId
     * @param tsKey
     */
    private void sendMessage2Successors(Object message, String parentId, Object tsKey) {
        Set<String> successors = processor2successors.get(parentId);
        if (successors != null) {
            for (String successor : successors) {
                logger.info("Send message to " + successor);
                collector.emit(TopologyStreamIds.DATAMESSAGE_STREAM, new Values(successor, message, tsKey, false));
            }
        } else {
            logger.debug("successors for " + parentId + " is null.");
            logger.debug("current processor map is " + new Gson().toJson(processor2successors));

        }
    }

    private String getProcessorId(DataMessage dataMessage, String posProcessorid) throws CPSException {
        if (dataMessage instanceof OptimizedParsedDataMessage) {
            return ((OptimizedParsedDataMessage) dataMessage).getSourceProcessorid();
        } else if (dataMessage instanceof ParsedDataMessage) {
            return posProcessorid;
        } else {
            throw new CPSException(ErrorCode.INVALID_INPUT_ERROR, "Error type of message.");
        }
    }

    private boolean idDeleteMetadataStream(String sourceStream) {
        if (sourceStream == null) {
            return false;
        }
        return sourceStream.equals(TopologyStreamIds.PROCESSOR_DELETE_SPOUT_STREAM);
    }

    private boolean isMetaDataStream(String sourceStream) {
        if (sourceStream == null) {
            return false;
        }
        return sourceStream.equals(TopologyStreamIds.PROCESSOR_SPOUT_STREAM);
    }

    private boolean isDataMessageStream(String sourceStream) {
        if (sourceStream == null) {
            return false;
        }
        return sourceStream.equals(TopologyStreamIds.PARSED_DATA_STREAM);
    }

    private boolean isBatchDataMessageStream(String sourceStream) {
        if (sourceStream == null) {
            return false;
        }
        return sourceStream.equals(TopologyStreamIds.STREAM_TO_DISPATCH);
    }

    @Override
    public void declareOutputFields(OutputFieldsDeclarer declarer) {
        declarer.declareStream(TopologyStreamIds.DATAMESSAGE_STREAM, new Fields(TopologyMessageFieldIds.DESTINATION,
                TopologyMessageFieldIds.MESSAGE, TopologyMessageFieldIds.TS_KEY, TopologyMessageFieldIds.IS_BUFFER));
        declarer.declareStream(TopologyStreamIds.METADATA_STREAM, new Fields(TopologyMessageFieldIds.MESSAGE));
        declarer.declareStream(TopologyStreamIds.METADATA_DELETE_STREAM,
                new Fields(TopologyMessageFieldIds.SOURCE, TopologyMessageFieldIds.DESTINATION));

    }
}
